package com.dog.modules.iot.mqtt.consumer;

import com.alibaba.fastjson2.JSON;
import com.dog.framework.core.constant.IotMqConstants;
import com.dog.framework.mqtt.MqttConstant;
import com.dog.framework.mqtt.component.IMqttListener;
import com.dog.framework.mqtt.config.MqttClientProperties;
import com.dog.framework.mqtt.domain.IotMessage;
import com.dog.modules.iot.mqtt.codec.BodyDecoder;
import com.dog.modules.iot.mqtt.consumer.bound.InboundCommand;
import com.dog.modules.iot.mqtt.consumer.bound.InboundHeader;
import com.dog.modules.iot.mqtt.consumer.enums.InCommandType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.UUID;


/**
 */
@Slf4j
@Component
public class RobotsMqttListener implements IMqttListener {

    public static final String CLEANING_DIRECT_EXCHANGE = "cleaning_direct_exchange";
    public static final String CLEANING_FANOUT_EXCHANGE = "cleaning_fanout_exchange";

    @Autowired
    private BodyDecoder inBodyParser;

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MqttClientProperties clientProperties;
    @Override
    public void consume(IotMessage message) {
        InboundCommand cmd = convertCmd(message);
        if (cmd != null) {
            sendMsg(cmd, JSON.toJSONString(cmd));
        } else {
            log.error("consume msg failed: {}", JSON.toJSONString(message));
        }
    }

    public void sendMsg(InboundCommand cmd, String content) {
        int mode = Integer.parseInt(cmd.getHeader().getDeviceCode().substring(0, 8)) % 10;
        log.info("queue {} consume msg for {} {} : {}", mode, cmd.getHeader().getDeviceCode(), InCommandType.valueOf(cmd.getHeader().getCmd()), content);
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(CLEANING_DIRECT_EXCHANGE, IotMqConstants.CLEANING_ROBOT_MSG_ROUTING_KEY_PREFIX + mode, content, correlationId);
    }


    private InboundCommand convertCmd(IotMessage message) {
        if (message == null) {
            log.error("invalid message.");
            return null;
        }
        String topic = message.getTopic();
        if (ObjectUtils.isEmpty(topic)) {
            log.error("invalid message, empty topic.");
            return null;
        }
        Integer cmd;
        String deviceCode;
        String mqttUuid = message.getHeaders().getId().toString();
        Long ts = null;
        if (topic.startsWith(MqttConstant.SYS_TOPIC_PREFIX) && topic.endsWith(MqttConstant.SYS_TOPIC_CLIENTS_DISCONNECTED_EVENT)) {
            // 下线消息
            cmd = InCommandType.DISCONNECT.getCmd();
            deviceCode = inBodyParser.parseUuidFromClientId(message.getBody());
            ts = System.currentTimeMillis();
        } else if (topic.startsWith(MqttConstant.SYS_TOPIC_PREFIX) && topic.endsWith(MqttConstant.SYS_TOPIC_CLIENTS_CONNECTED_EVENT)) {
            // 上线消息
            cmd = InCommandType.CONNECT.getCmd();
            deviceCode = inBodyParser.parseUuidFromClientId(message.getBody());
            ts = System.currentTimeMillis();
        } else {
            // 普通业务消息
            InboundHeader header = inBodyParser.parseHeader(message.getBody());
            cmd = header.getCmd();
            deviceCode = header.getDeviceCode();
            ts = header.getTs();
        }
        if (ObjectUtils.isEmpty(deviceCode)) {
            log.info("invalid message,empty uuid.");
            return null;
        }
        //服务客户端连接到emqx
        if (deviceCode.contains(clientProperties.getId())) {
            log.info("server[{}] is connected to emqx broker.", deviceCode);
            return null;
        }
        InboundCommand inboundCommand = new InboundCommand();
        InboundHeader header = new InboundHeader(deviceCode, cmd, mqttUuid, ts);
        inboundCommand.setHeader(header);
        inboundCommand.setData(inBodyParser.parseData(message.getBody()));
        return inboundCommand;
    }

}
